#include "rpc_probe.h"
#include "rpc_proxy.h"
#include "rpc_root.h"
#include "rpc_statistics_impl.hh"
#include "rpc_stub.h"
#include <stdexcept>

#if defined(_MSC_VER)
#include <Windows.h>
#pragma comment(lib, "wsock32.lib")
#endif

namespace rpc {

const char *getErrorString(RpcError error) {
  switch (error) {
  case RpcError::EXCEPTION:
    return "RPC Service exception";
  case RpcError::NOT_FOUND:
    return "RPC service or method not found";
  case RpcError::TIMEOUT:
    return "RPC service's method timed out";
  case RpcError::OK:
    return "ok";
  default:
    return "";
  }
  return "";
}

RpcImpl::RpcImpl() {}

RpcImpl::~RpcImpl() {}

void RpcImpl::initialize(Logger *logger, bool coroMode, bool openStat) {
  if (!rpcMalloc_) {
    rpcMalloc_ = std::malloc;
  }
  if (!rpcFree_) {
    rpcFree_ = std::free;
  }

  logger_ = logger;
  coroMode_ = coroMode;
  openStat_ = openStat;

  proxyManager_ = allocObject<ProxyManagerImpl>(this);
  stubManager_ = allocObject<StubManagerImpl>(this);
  proxyCallManager_ = allocObject<ProxyCallManagerImpl>(this);
  stubCallManager_ = allocObject<StubCallManagerImpl>(this);
}

void RpcImpl::deinitialize() {
  if (openStat_) {
    uninitializeStatistics();
  }
  if (proxyCallManager_) {
    proxyCallManager_->cleanup();
  }
  if (stubCallManager_) {
    stubCallManager_->cleanup();
  }
  if (stubManager_) {
    stubManager_->cleanup();
  }
  if (proxyManager_) {
    proxyManager_->cleanup();
  }
  deallocObject(proxyCallManager_);
  deallocObject(stubCallManager_);
  deallocObject(proxyManager_);
  deallocObject(stubManager_);
  if (rpcProbe_) {
    rpcProbe_->on_close();
  }
}

void RpcImpl::setAllocator(RpcMalloc mallocFunc, RpcFree freeFunc) {
  rpcMalloc_ = mallocFunc;
  rpcFree_ = freeFunc;
}

void RpcImpl::update(std::time_t ms) {
  coro_sched();
  if (proxyCallManager_) {
    proxyCallManager_->update(ms);
  }
  if (stubCallManager_) {
    stubCallManager_->update(ms);
  }
  if (stubManager_) {
    stubManager_->update(ms);
  }
  if (rpcProbe_) {
    rpcProbe_->on_tick(ms);
  }
}

bool RpcImpl::readHeader(TransportPtr &transport, RpcMsgHeader &header) {
  if (transport->size() <= static_cast<int>(sizeof(RpcMsgHeader))) {
    // No enough byte in buffer
    return false;
  }
  transport->peek(reinterpret_cast<char *>(&header), sizeof(header));
  header.ntoh();
  if (header.length == 0) {
    return false;
  }
  if ((int)(header.length) > transport->size()) {
    // No enough byte in buffer
    return false;
  }
  switch (static_cast<RpcMsgType>(header.type)) {
  case RpcMsgType::RPC_CALL:
  case RpcMsgType::RPC_RETURN:
  case RpcMsgType::RPC_PROXY_CALL:
  case RpcMsgType::RPC_PROXY_RETURN:
  case RpcMsgType::NO_RPC:
    break;
  default:
    return false;
  }
  return true;
}

void RpcImpl::doMessage(const RpcMsgHeader &header, TransportPtr &transport,
                        kratos::service::ServiceContext *ctx,
                        rpc::ProxyHandler *proxyHandler,
                        rpc::NoRpcHander *nonRpcHandler) {
  auto msgType = static_cast<RpcMsgType>(header.type);
  if (proxyHandler) {
    // Proxy, user should transfer to internal grid
    proxyHandler->onRelay(transport, header);
  } else {
    switch (msgType) {
    case RpcMsgType::RPC_CALL:
      onCall(header, transport, ctx);
      break;
    case RpcMsgType::RPC_RETURN:
      onReturn(header, transport);
      break;
    case RpcMsgType::RPC_PROXY_CALL:
      onProxyCall(header, transport, ctx);
      break;
    case RpcMsgType::RPC_PROXY_RETURN:
      onProxyCallReturn(header, transport);
      break;
    case RpcMsgType::NO_RPC:
      // non-RPC message
      if (nonRpcHandler) {
        nonRpcHandler->onMessage(transport, header);
      }
      break;
    default:
      break;
    }
  }
}

void RpcImpl::onMessage(TransportPtr &transport,
                        kratos::service::ServiceContext *ctx,
                        rpc::ProxyHandler *proxyHandler,
                        rpc::NoRpcHander *nonRpcHandler) {
  static RpcMsgHeader header;
  while (true) {
    if (!transport || transport->isClose()) {
      return;
    }
    if (!readHeader(transport, header)) {
      return;
    }
    if (header.length == 0) {
      transport->close();
      return;
    }
    auto beforeSize = transport->size();
    if (rpcProbe_) {
      rpcProbe_->on_msg((RpcMsgType)(header.type), transport);
    }
    doMessage(header, transport, ctx, proxyHandler, nonRpcHandler);
    auto actualConsume = beforeSize - transport->size();
    if (actualConsume != static_cast<int>(header.length)) {
      if (actualConsume > static_cast<int>(header.length)) {
        transport->close();
      } else {
        transport->skip(header.length - actualConsume);
      }
    }
  }
}

StubPtr RpcImpl::getService(ServiceID serviceID, ServiceUUID uuid,
                            kratos::service::ServiceContext *ctx) {
  if (serviceID) {
    // Use specific service
    auto stub = StubCreatorRef.getService(this, serviceID);
    if (!stub) {
      // Query/Create a new service
      return StubCreatorRef.createService(this, uuid, ctx);
    } else {
      return stub;
    }
  } else {
    // Query/Create a new service
    return StubCreatorRef.createService(this, uuid, ctx);
  }
}

InterfaceDescriptorFactory *RpcImpl::getDescriptorFactory() {
  return descriptorFactory_;
}

void RpcImpl::setRpcProbe(RpcProbe *probe) { rpcProbe_ = probe; }

RpcProbe *RpcImpl::getRpcProbe() { return rpcProbe_; }

void RpcImpl::onCall(const RpcMsgHeader &header, TransportPtr &transport,
                     kratos::service::ServiceContext *ctx) {
  // Receive a call header
  static RpcCallHeader callHeader;
  transport->recv(reinterpret_cast<char *>(&callHeader), sizeof(callHeader));
  callHeader.ntoh();
  // Query service
  StubPtr stub = getService(callHeader.callHeader.serviceID,
                            callHeader.callHeader.serviceUUID, ctx);
  if (!stub) {
    notFound(transport, callHeader.callHeader.callID);
    return;
  }
  auto stubImpl = std::dynamic_pointer_cast<StubImpl>(stub);
  if (!stubImpl) {
    // Never reach
    return;
  }
  // Calculate body length
  auto bytes = header.length - sizeof(callHeader);
  // Call service
  stubImpl->callStub(transport, static_cast<int>(bytes), callHeader.callHeader);
}

void RpcImpl::onProxyCall(const RpcMsgHeader &header, TransportPtr &transport,
                          kratos::service::ServiceContext *ctx) {
  static RpcProxyCallHeader callHeader;
  transport->recv(reinterpret_cast<char *>(&callHeader), sizeof(callHeader));
  callHeader.ntoh();
  StubPtr stub = getService(callHeader.callHeader.serviceID,
                            callHeader.callHeader.serviceUUID, ctx);
  if (!stub) {
    notFound(transport, callHeader.callHeader.callID);
    return;
  }
  auto stubImpl = std::dynamic_pointer_cast<StubImpl>(stub);
  if (!stubImpl) {
    // Never reach
    return;
  }
  // Calculate body length
  auto bytes = header.length - sizeof(callHeader);
  // Call service
  stubImpl->callStubProxy(transport, static_cast<int>(bytes),
                          callHeader.callHeader);
}

void RpcImpl::onReturn(const RpcMsgHeader &header, TransportPtr &transport) {
  static RpcRetHeader retHeader;
  transport->recv(reinterpret_cast<char *>(&retHeader), sizeof(RpcRetHeader));
  retHeader.ntoh();
  auto call = proxyCallManager_->get(retHeader.retHeader.callID);
  if (!call) {
    Logbegin << "ProxyCall not found [" << retHeader.retHeader.callID << "]"
             << Logend;
    return;
  }
  call->setErrorID(retHeader.retHeader.errorID);
  auto proxy = getProxyManager()->get(call->getProxyID());
  if (!proxy) {
    Logbegin << "Proxy not found [" << call->getProxyID() << "]" << Logend;
    getProxyCallManager()->destroy(retHeader.retHeader.callID);
    return;
  }
  auto proxyImpl = std::dynamic_pointer_cast<ProxyImpl>(proxy);
  auto bytes = header.length - sizeof(retHeader);
  auto errorID = static_cast<RpcError>(retHeader.retHeader.errorID);
  switch (errorID) {
  case RpcError::NOT_FOUND:
    Logbegin << "Remote service or method not found [" << call->getCallName()
             << "]" << Logend;
    proxyImpl->setTarget(INVALID_SERVICE_ID);
    break;
  case RpcError::EXCEPTION:
    Logbegin << "Remote service's method thrown exception ["
             << call->getCallName() << "]" << Logend;
    proxyImpl->setTarget(INVALID_SERVICE_ID);
    break;
  case RpcError::OK:
    if (proxyImpl->isAttached()) {
      // Need to attach the service
      proxyImpl->setTarget(retHeader.retHeader.serviceID);
    } else {
      proxyImpl->setTarget(INVALID_SERVICE_ID);
    }
    break;
  default:
    proxyImpl->setTarget(INVALID_SERVICE_ID);
    Logbegin << "Unknown RPC error " << retHeader.retHeader.errorID << " ["
             << call->getCallName() << "]" << Logend;
    getProxyCallManager()->destroy(call->getCallID());
    return;
  }
  call->doRet(static_cast<int>(bytes));
}

void RpcImpl::onProxyCallReturn(const RpcMsgHeader &header,
                                TransportPtr &transport) {
  static RpcProxyRetHeader retHeader;
  transport->recv(reinterpret_cast<char *>(&retHeader),
                  sizeof(RpcProxyRetHeader));
  retHeader.ntoh();
  auto call = proxyCallManager_->get(retHeader.retHeader.callID);
  if (!call) {
    Logbegin << "ProxyCall not found [" << retHeader.retHeader.callID << "]"
             << Logend;
    return;
  }
  call->setErrorID(retHeader.retHeader.errorID);
  auto proxy = proxyManager_->get(call->getProxyID());
  if (!proxy) {
    Logbegin << "Proxy not found [" << call->getProxyID() << "]" << Logend;
    proxyCallManager_->destroy(retHeader.retHeader.callID);
    return;
  }
  auto proxyImpl = std::dynamic_pointer_cast<ProxyImpl>(proxy);
  auto bytes = header.length - sizeof(retHeader);
  auto errorID = static_cast<RpcError>(retHeader.retHeader.errorID);
  switch (errorID) {
  case RpcError::NOT_FOUND:
    Logbegin << "Remote service or method not found [" << call->getCallName()
             << "]" << Logend;
    proxyImpl->setTarget(INVALID_SERVICE_ID);
    break;
  case RpcError::EXCEPTION:
    Logbegin << "Remote service's method thrown exception ["
             << call->getCallName() << "]" << Logend;
    proxyImpl->setTarget(INVALID_SERVICE_ID);
    break;
  case RpcError::OK:
    if (proxyImpl->isAttached()) {
      proxyImpl->setTarget(retHeader.retHeader.serviceID);
    } else {
      proxyImpl->setTarget(INVALID_SERVICE_ID);
    }
    break;
  default:
    proxyImpl->setTarget(INVALID_SERVICE_ID);
    Logbegin << "Unknown RPC error " << retHeader.retHeader.errorID << " ["
             << call->getCallName() << "]" << Logend;
    proxyCallManager_->destroy(call->getCallID());
    return;
  }
  call->doRet(static_cast<int>(bytes));
}

void RpcImpl::notFound(TransportPtr &transport, CallID callID) {
  static RpcRetHeader ret;
  ret.header.length = sizeof(ret);
  ret.header.type = static_cast<MsgTypeID>(RpcMsgType::RPC_RETURN);
  ret.retHeader.callID = callID;
  ret.retHeader.errorID = static_cast<ErrorID>(RpcError::NOT_FOUND);
  ret.hton();
  transport->send(reinterpret_cast<char *>(&ret), sizeof(ret));
}

Logger *RpcImpl::getLogger() { return logger_; }

void RpcImpl::setContext(void *context) { context_ = context; }

/**
 * Relay buffer.
 */
class RelayBuffer {
  std::size_t size_{0};   ///< Total size of buffer
  std::size_t length_{0}; ///< Current occupied size
  char *buffer_{nullptr}; ///< Buffer memory
  Rpc *rpc_{nullptr};

public:
  /**
   * ctor.
   *
   * \param size Maximum size of buffer
   */
  RelayBuffer(Rpc *rpc, std::size_t size) {
    rpc_ = rpc;
    reset(size);
  }
  /**
   * dtor.
   *
   */
  ~RelayBuffer() {}
  /**
   * Relay and transform proxy call to a normal call.
   *
   * \param from Source
   * \param to Destination
   * \param header RpcMsgHeader
   * \return true or false
   */
  bool doCallHeader(TransportPtr &from, TransportPtr &to,
                    const RpcMsgHeader &header, CallID fake_call_id) {
    auto bodyLength = header.length - sizeof(RpcProxyCallHeader);
    reset(bodyLength + sizeof(RpcCallHeader));
    auto *callHeader = reinterpret_cast<RpcCallHeader *>(buffer_);
    callHeader->header.type =
        (std::underlying_type<RpcMsgType>::type)RpcMsgType::RPC_CALL;
    callHeader->header.length = bodyLength + sizeof(RpcCallHeader);
    static RpcProxyCallRequestHeader proxyCallRequestHeader;
    from->skip(sizeof(RpcMsgHeader));
    from->peek(reinterpret_cast<char *>(&proxyCallRequestHeader),
               sizeof(RpcProxyCallRequestHeader));
    proxyCallRequestHeader.ntoh();
    callHeader->callHeader.serviceUUID = proxyCallRequestHeader.serviceUUID;
    callHeader->callHeader.serviceID = proxyCallRequestHeader.serviceID;
    callHeader->callHeader.methodID = proxyCallRequestHeader.methodID;
    if (fake_call_id != INVALID_CALL_ID) {
      callHeader->callHeader.callID = fake_call_id;
    } else {
      callHeader->callHeader.callID = proxyCallRequestHeader.callID;
    }
    from->skip(sizeof(RpcProxyCallRequestHeader));
    callHeader->hton();
    length_ = sizeof(RpcCallHeader);
    return relay(from, to, bodyLength);
  }
  /**
   * Relay and transform proxy call return to a normal call return.
   *
   * \param from Source
   * \param to Destination
   * \param header RpcMsgHeader
   * \return true or false
   */
  bool doCallRetHeader(TransportPtr &from, TransportPtr &to,
                       const RpcMsgHeader &header) {
    auto bodyLength = header.length - sizeof(RpcProxyRetHeader);
    reset(bodyLength + sizeof(RpcRetHeader));
    auto *callRetHeader = reinterpret_cast<RpcRetHeader *>(buffer_);
    callRetHeader->header.type =
        (std::underlying_type<RpcMsgType>::type)RpcMsgType::RPC_RETURN;
    callRetHeader->header.length = bodyLength + sizeof(RpcRetHeader);
    static RpcProxyCallRetHeader proxyCallRetHeader;
    from->skip(sizeof(RpcMsgHeader));
    from->peek(reinterpret_cast<char *>(&proxyCallRetHeader),
               sizeof(RpcProxyCallRetHeader));
    proxyCallRetHeader.ntoh();
    callRetHeader->retHeader.callID = proxyCallRetHeader.callID;
    callRetHeader->retHeader.errorID = proxyCallRetHeader.errorID;
    callRetHeader->retHeader.serviceID = proxyCallRetHeader.serviceID;
    from->skip(sizeof(RpcProxyCallRetHeader));
    callRetHeader->hton();
    length_ = sizeof(RpcRetHeader);
    return relay(from, to, bodyLength);
  }
  /**
   * Relay and transform normal call to a proxy call.
   *
   * \param from Source
   * \param to Destination
   * \param header RpcMsgHeader
   * \return true or false
   */
  bool doProxyCallHeader(TransportPtr &from, TransportPtr &to,
                         const RpcMsgHeader &header) {
    if (from->getGlobalIndex() == INVALID_GLOBAL_INDEX) {
      return false;
    }
    auto bodyLength = header.length - sizeof(RpcCallHeader);
    reset(bodyLength + sizeof(RpcProxyCallHeader));
    auto *proxyCallHeader = reinterpret_cast<RpcProxyCallHeader *>(buffer_);
    proxyCallHeader->header.type =
        (std::underlying_type<RpcMsgType>::type)RpcMsgType::RPC_PROXY_CALL;
    proxyCallHeader->header.length = bodyLength + sizeof(RpcProxyCallHeader);
    static RpcCallRequestHeader requestHeader;
    from->skip(sizeof(RpcMsgHeader));
    from->peek(reinterpret_cast<char *>(&requestHeader),
               sizeof(RpcCallRequestHeader));
    requestHeader.ntoh();
    proxyCallHeader->callHeader.serviceUUID = requestHeader.serviceUUID;
    proxyCallHeader->callHeader.serviceID = requestHeader.serviceID;
    proxyCallHeader->callHeader.methodID = requestHeader.methodID;
    proxyCallHeader->callHeader.globalIndex = from->getGlobalIndex();
    proxyCallHeader->callHeader.callID = requestHeader.callID;
    from->skip(sizeof(RpcCallRequestHeader));
    proxyCallHeader->hton();
    length_ = sizeof(RpcProxyCallHeader);
    return relay(from, to, bodyLength);
  }
  /**
   * Relay and transform normal call return to a proxy call return.
   *
   * \param from Source
   * \param to Destination
   * \param header RpcMsgHeader
   * \return true or false
   */
  bool doProxyRetHeader(TransportPtr &from, TransportPtr &to,
                        const RpcMsgHeader &header, rpc::CallID real_call_id) {
    if (from->getGlobalIndex() == INVALID_GLOBAL_INDEX) {
      return false;
    }
    auto bodyLength = header.length - sizeof(RpcRetHeader);
    reset(bodyLength + sizeof(RpcProxyRetHeader));
    auto *proxyRetHeader = reinterpret_cast<RpcProxyRetHeader *>(buffer_);
    proxyRetHeader->header.type =
        (std::underlying_type<RpcMsgType>::type)RpcMsgType::RPC_PROXY_RETURN;
    proxyRetHeader->header.length = bodyLength + sizeof(RpcProxyRetHeader);
    static RpcCallRetHeader retHeader;
    from->skip(sizeof(RpcMsgHeader));
    from->peek(reinterpret_cast<char *>(&retHeader), sizeof(RpcCallRetHeader));
    retHeader.ntoh();
    proxyRetHeader->retHeader.callID = real_call_id;
    proxyRetHeader->retHeader.errorID = retHeader.errorID;
    proxyRetHeader->retHeader.serviceID = retHeader.serviceID;
    proxyRetHeader->retHeader.globalIndex = from->getGlobalIndex();
    from->skip(sizeof(RpcCallRetHeader));
    proxyRetHeader->hton();
    length_ = sizeof(RpcProxyRetHeader);
    return relay(from, to, bodyLength);
  }
  /**
   * Resend to destination.
   *
   * \param from Source
   * \param to Destination
   * \return true or false
   */
  bool relay(TransportPtr &from, TransportPtr &to, std::uint32_t bodyLength) {
    if (from->isClose() || to->isClose()) {
      return false;
    }
    from->recv(buffer_ + length_, bodyLength);
    to->send(buffer_, length_ + bodyLength);
    return true;
  }

private:
  /**
   * Reset buffer to new size.
   *
   * \param size new size of buffer
   */
  void reset(std::size_t size) {
    length_ = 0;
    if (!buffer_) {
      size_ = size;
      buffer_ = (char *)rpc_->malloc(size);
    } else {
      if (size > size_) {
        size_ = size;
        rpc_->free(buffer_);
        buffer_ = (char *)rpc_->malloc(size);
      }
    }
  }
};

bool RpcImpl::relay(TransportPtr &from, TransportPtr &to,
                    const RpcMsgHeader &header, CallID replace_call_id) {
  static RelayBuffer relayBuffer(this, STREAMBUF_SIZE);
  if (from->isClose() || to->isClose()) {
    return false;
  }
  auto msgType = (RpcMsgType)header.type;
  if (msgType == RpcMsgType::RPC_CALL) {
    // transfer call to proxy call
    relayBuffer.doProxyCallHeader(from, to, header);
  } else if (msgType == RpcMsgType::RPC_RETURN) {
    // transfer call return to proxy call return
    relayBuffer.doProxyRetHeader(from, to, header, replace_call_id);
  } else if (msgType == RpcMsgType::RPC_PROXY_CALL) {
    // transfer proxy call to call
    relayBuffer.doCallHeader(from, to, header, replace_call_id);
  } else if (msgType == RpcMsgType::RPC_PROXY_RETURN) {
    // transfer proxy call return to call return
    relayBuffer.doCallRetHeader(from, to, header);
  } else {
    return false;
  }
  return true;
}

ProxyManager *RpcImpl::getProxyManager() { return proxyManager_; }

StubManager *RpcImpl::getStubManager() { return stubManager_; }

ProxyCallManager *RpcImpl::getProxyCallManager() { return proxyCallManager_; }

StubCallManager *RpcImpl::getStubCallManager() { return stubCallManager_; }

void *RpcImpl::malloc(std::size_t size) { return rpcMalloc_(size); }
void RpcImpl::free(void *p) { return rpcFree_(p); }

ProxyCreator *RpcImpl::getProxyCreator() {
  return &rpc::getGlobalProxyCreator();
}

bool RpcImpl::switchRpcStat(bool on) {
  auto ret = openStat_;
  openStat_ = on;
  return ret;
}

void RpcImpl::setDescriptorFactory(
    InterfaceDescriptorFactory *descriptorFactory) {
  descriptorFactory_ = descriptorFactory;
}

void *RpcImpl::getContextRaw() { return context_; }

void buildHeader(RpcCallHeader &cheader, rpc::Proxy *proxy, rpc::CallID callID,
                 rpc::MethodID methodID, std::uint32_t bodySize) {
  cheader.header.length = sizeof(RpcCallHeader) + bodySize;
  cheader.header.type = static_cast<rpc::MsgTypeID>(rpc::RpcMsgType::RPC_CALL);
  cheader.callHeader.serviceUUID = proxy->getServiceUUID();
  cheader.callHeader.serviceID = dynamic_cast<ProxyImpl *>(proxy)->getTarget();
  cheader.callHeader.methodID = methodID;
  cheader.callHeader.callID = callID;
  cheader.hton();
}

void buildHeader(RpcCallHeader &cheader, rpc::ServiceUUID service_uuid,
                 rpc::ServiceID service_id, rpc::CallID callID,
                 rpc::MethodID methodID, std::uint32_t bodySize) {
  cheader.header.length = sizeof(RpcCallHeader) + bodySize;
  cheader.header.type = static_cast<rpc::MsgTypeID>(rpc::RpcMsgType::RPC_CALL);
  cheader.callHeader.serviceUUID = service_uuid;
  cheader.callHeader.serviceID = service_id;
  cheader.callHeader.methodID = methodID;
  cheader.callHeader.callID = callID;
  cheader.hton();
}

void buildHeader(RpcProxyCallHeader &cheader, rpc::Proxy *proxy,
                 rpc::CallID callID, rpc::MethodID methodID,
                 std::uint32_t bodySize, bool oneway) {
  cheader.header.length = sizeof(RpcProxyCallHeader) + bodySize;
  cheader.header.type =
      static_cast<rpc::MsgTypeID>(rpc::RpcMsgType::RPC_PROXY_CALL);
  cheader.callHeader.serviceUUID = proxy->getServiceUUID();
  cheader.callHeader.serviceID = dynamic_cast<ProxyImpl *>(proxy)->getTarget();
  cheader.callHeader.methodID = methodID;
  cheader.callHeader.callID = callID;
  cheader.callHeader.globalIndex =
      dynamic_cast<ProxyImpl *>(proxy)->getGlobalIndex();
  cheader.callHeader.oneWay = oneway;
  cheader.hton();
}

LogWriter::LogWriter(Logger *logger) : logger_(logger) { ss_ << "[RPC]"; }

bool loadClass(Rpc *rpc, const std::string &uuid, const std::string &bundlePath,
               bool noCreate, kratos::service::ServiceContext *ctx) {
  if (uuid.empty() || bundlePath.empty()) {
    return false;
  }

  auto result = StubCreatorRef.loadClass(rpc, std::stoull(uuid), bundlePath,
                                         noCreate, ctx);
  if (rpc->getRpcProbe()) {
    rpc->getRpcProbe()->on_load(uuid, bundlePath, result);
  }
  return result;
}

bool destroyServiceUnsafe(Rpc *rpc, const std::string &uuid) {
  if (!rpc->getStubManager()) {
    return true;
  }
  auto &serviceList =
      rpc->getStubManager()->getMutableServiceMapUnsafe(std::stoull(uuid));
  if (serviceList.empty()) {
    return true;
  }
  for (auto &service : serviceList) {
    Logbegin << "Destroy service [" << service->getUUID() << "]["
             << service->getServiceName() << "]" << Logend;
    service->reset(nullptr, rpc);
  }
  rpc->getStubManager()->removeAllServiceByUUID(std::stoull(uuid));
  serviceList.clear();
  return true;
}

bool unloadClassUnsafe(Rpc *rpc, const std::string &uuid) {
  if (!destroyServiceUnsafe(rpc, uuid)) {
    return false;
  }
  auto result = StubCreatorRef.unloadClass(std::stoull(uuid));
  if (rpc->getRpcProbe()) {
    rpc->getRpcProbe()->on_unload(uuid, result);
  }
  return result;
}

void coro(coroutine::CoroEntry coro) { coro_start(coro); }

void deinitialize() {
  coro_shutdown();
  ObjectPoolManagerSinglton::destroy();
}

} // namespace rpc
